package ru.nagakhl.whirlpool;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.Cassandra.Client;

import ru.nagakhl.whirlpool.exceptions.ConnectorException;
import ru.nagakhl.whirlpool.exceptions.PoolException;
import ru.nagakhl.whirlpool.pool.BlockingConnectionPool;
import ru.nagakhl.whirlpool.pool.SimplePool;

public class ConnectionManagerImpl implements ConnectionManager {

	private static final String MAX_CONNECTIONS = "max_connections";

	private static final String INITIAL_CONNECTIONS = "initial_connections";

	private static final String CASSANDRA_NODES = "cassandra_nodes";

	private static final String ALLOCATION_STRATEGY = "allocation_strategy";
//	
//	private static ConsistencyLevel readConsistency;
//	
//	private static ConsistencyLevel writeConsistency;

	protected final int initial;

	protected final int max;

	protected String activeHost;
	
	protected Map<String, BlockingConnectionPool> pools = new HashMap<String, BlockingConnectionPool>();
	
	protected Map<Class<?>, BasicMetadata> metadata = new HashMap<Class<?>, BasicMetadata>();
	
	protected final Lock poolLock;
	
	protected final Lock metadataLock;

	public ConnectionManagerImpl(Properties props) {
		initial = Integer.parseInt(props.getProperty(INITIAL_CONNECTIONS, "1"));
		max = Integer.parseInt(props.getProperty(MAX_CONNECTIONS, "10"));
		poolLock = new ReentrantLock();
		metadataLock = new ReentrantLock();
		String strategy = props.getProperty(ALLOCATION_STRATEGY, "ALL_FIRST");
		String[] hosts = props.getProperty(CASSANDRA_NODES, "127.0.0.1:9160").split(",");
		if ("ALL_FIRST".equals(strategy)) {
			initAllFirst(hosts);
		} else if ("UNIFORM".equals(strategy)) {
			initUniform(hosts);
		} else {
			throw new IllegalArgumentException("Unknown connection distribution strategy.");
		}
	}

	public Connection createConnection() {
		final BlockingConnectionPool activePool;
		poolLock.lock();
		try {
			activePool = pools.get(activeHost);
		} finally {
			poolLock.unlock();
		}
		Connection con = new ConnectionImpl(activePool.getConnection(), new ManagerConnector() {
			public void releaseClient(Client client) {
				activePool.returnConnection(client);
			}

			public void notifyFailure(Exception ex) {
				handleError(ex);
			}

			public Metadata getMetadata(Class<?> cls) {
				return getClassMetadata(cls);
			}
		});
		return con;
	}

	protected void initAllFirst(String[] hosts) {
		if (hosts.length > 0) {
			for (String host : hosts) {
				String[] desc = host.split(":");
				int port;
				if (desc.length > 1)
					port = Integer.parseInt(desc[1]);
				else
					port = 9160;
				pools.put(host, new PoolEnvelope(initial, max, port, desc[0]));
			}
			activeHost = hosts[0];
			BlockingConnectionPool first = pools.get(hosts[0]);
			Cassandra.Client cl = first.getConnection();
			first.returnConnection(cl);
		} else {
			throw new ConnectorException("No cassandra hosts found in configuration");
		}
	}

	protected void initUniform(String[] hosts) {
		throw new UnsupportedOperationException("Not implemented!");
	}

	protected void rebuildPool() {
		poolLock.lock();
		try {
			throw new UnsupportedOperationException("Not yet implemented!");
		} finally {
			poolLock.unlock();
		}
	}
	
	protected Metadata getClassMetadata(Class<?> cls) {
		BasicMetadata m;
		if (metadata.containsKey(cls)) {
			m = metadata.get(cls);
		} else {
			metadataLock.lock();
			try {
				m = new BasicMetadata(cls, ConsistencyLevel.QUORUM);
				metadata.put(cls, m);
			} finally {
				metadataLock.unlock();
			}
		}
		return m;
	}
	
	protected void handleError(Exception ex) {
		throw new UnsupportedOperationException("Not yet implemented!");
	}
	
		private class PoolEnvelope implements BlockingConnectionPool {

		BlockingConnectionPool realPool = null;

		protected int initial;

		protected int max;

		protected int port;

		protected String host;

		public PoolEnvelope(int initial, int max, int port, String host) {
			this.initial = initial;
			this.max = max;
			this.port = port;
			this.host = host;
		}

		public void returnConnection(Client cl) throws PoolException {
			if (realPool != null) {
				realPool.returnConnection(cl);
			} else {
				throw new ConnectorException("Attempt to return connection into uninitialized pool.");
			}
		}

		public Client getConnection() {
			if (realPool == null) {
				realPool = new SimplePool(initial, max, host, port);
			}
			return realPool.getConnection();
		}

	}

}
